#include "CBL_ThreadPool.h"
#include "CBL_Cpu.h"
#include "CBL_Profile.h"

CB_SUBNAMESPACE_START(Cpu)

// -------------------------------------------------------------------------
//	CJobProcess
// -------------------------------------------------------------------------

void CThreadPool::CJobProcess::AssignJob(CDelegate& _rJob)
{
	CB_CRITICAL_SECTION_GUARD(m_WorkMutex);

	if (Debug::Verify(!m_Working, "Cannot assign a new job to a working jobprocess."))
	{
		m_Job     = _rJob;
		m_Working = true;
	}
}

// -------------------------------------------------------------------------

bool CThreadPool::CJobProcess::IsAvailable()
{
	CB_CRITICAL_SECTION_GUARD(m_WorkMutex);
	return !m_Working;
}

// -------------------------------------------------------------------------

void CThreadPool::CJobProcess::Tick(void* CB_UNUSED(_pData))
{
	CB_CRITICAL_SECTION_GUARD(m_WorkMutex);
	if (m_Working)
	{
		m_Job();
		m_Working = false;
	}
}

// -------------------------------------------------------------------------
//	CThreadPool
// -------------------------------------------------------------------------

void CThreadPool::PushJob(CDelegate& _rJob)
{
	CB_PROFILE("CThreadPool::PushJob");
    CB_CRITICAL_SECTION_START(m_DrainMutex);
    CB_CRITICAL_SECTION_START(m_QueueMutex);
    
    m_JobQueue.push_front(_rJob);
    
    CB_CRITICAL_SECTION_END;
    CB_CRITICAL_SECTION_END;
    
    Resume();
}

// -------------------------------------------------------------------------

void CThreadPool::Drain()
{
    CB_CRITICAL_SECTION_GUARD(m_DrainMutex);

	// Empty the queue, block push
    while (CriticalSectionTest(m_QueueMutex, !m_JobQueue.empty()))
		Wait(10);

	// Queue is empty, but there are still threads running
	bool JobsActive = true;
	while (JobsActive)
	{
		JobsActive = false;
		for (uint32 Slot(0); Slot < m_PoolSize; ++Slot)
		{
			if (!m_pThreadPool[Slot].IsAvailable())
			{
				JobsActive = true;
				break;
			}
		}
		Wait(10);
	}    
}

// -------------------------------------------------------------------------

void CThreadPool::Tick(void* CB_UNUSED(_pData))
{
	CB_PROFILE("CThreadPool::Tick");
	CB_CRITICAL_SECTION_GUARD(m_QueueMutex);
    
    for (uint32 Slot(0); (Slot < m_PoolSize) && !m_JobQueue.empty(); ++Slot)
    {
        if (m_pThreadPool[Slot].IsAvailable())
        {
        	m_pThreadPool[Slot].AssignJob(m_JobQueue.back());
			m_JobQueue.pop_back();
        }
    }
}

// -------------------------------------------------------------------------

CThreadPool::CThreadPool()
	: CThread("CThreadPool", 10)
{
	m_PoolSize = Cpu::GetTotalNumberOfProcessors();
    m_pThreadPool = new CJobProcess[m_PoolSize];

	for (uint32 Slot(0); Slot < m_PoolSize; ++Slot)
    	m_pThreadPool[Slot].Run();
	
    Run();
}

// -------------------------------------------------------------------------

CThreadPool::~CThreadPool()
{
	Stop();
    delete[] m_pThreadPool;
}

CB_SUBNAMESPACE_END